package com.itcast.flink.connectors.jdbc;

import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.List;

/**
 * <p>Description: </p>
 *
 * @author
 * @version 1.0
 * <p>Copyright:Copyright(c)2020</p>
 * @date
 */
public class JdbcConncetorsApplication {
    
    public static void main(String[] args) throws Exception {
        // 1.创建运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 2. 创建集合数据
        List arrs = new ArrayList<String>();
        arrs.add("10.10.20.10\t1603166893313\tPOST\taddOrder");
        arrs.add("10.10.20.11\t1603166893313\tGET\tgetOrder");
        // 3. 读取集合数据， 写入数据库
        env.fromCollection(arrs).addSink(JdbcSink.sink(
                // 配置SQL语句
                "insert into t_access_log (ip, time, type, api) values(?, ?, ?, ?)",
                (ps, value) -> {
                    System.out.println("receive ==> " + value);
                    // 解析数据
                    String[] arrValue = String.valueOf(value).split("\t");
                    for (int i = 0; i < arrValue.length; i++) {
                        // 新增数据
                        ps.setString(i + 1, arrValue[i]);
                    }
                    
                },
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:mysql://192.168.23.140:3306/flink?useSSL=false")
                        .withDriverName("com.mysql.jdbc.Driver")
                        .withUsername("root")
                        .withPassword("654321")
                        .build()
        ));
        
        // 4. 执行任务
        env.execute("jdbc sink job");
    }
}
